package crp.core.resource.domain.sync;

import com.google.common.base.Throwables;
import com.google.common.collect.Sets;
import com.qwlabs.lang.QueueWorker;
import crp.core.resource.domain.events.SyncResourceFired;
import crp.core.resource.domain.models.TimetableResourceSyncPriorities;
import crp.core.resource.domain.models.TimetableResourceSyncRecordEntity;
import crp.core.resource.domain.repositories.TimetableResourceSyncRecordRepository;
import crp.core.shared.ProcessStatus;
import com.qwlabs.cdi.DispatchInstance;
import crp.core.shared.config.AppEnv;
import crp.core.shared.utils.Times;
import io.quarkus.vertx.ConsumeEvent;
import lombok.extern.slf4j.Slf4j;

import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.inject.Instance;
import javax.inject.Inject;
import javax.transaction.Transactional;
import java.time.Duration;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@ApplicationScoped
public class ResourceSynchronizer {
    private static final Duration SPIN_DURATION = Duration.ZERO;
    private static final Duration TIMEOUT = Duration.ofMinutes(10);
    private static final Duration ENFORCE_SUCCEED_DURATION = Duration.ofDays(7);
    private final DispatchInstance<String, TimetableResourceSynchronizer> dispatcher;
    private final TimetableResourceSyncRecordRepository syncRecordRepository;
    private final AppEnv appEnv;
    private final Times times;
    private final AtomicBoolean running;

    @Inject
    public ResourceSynchronizer(Instance<TimetableResourceSynchronizer> instance,
                                TimetableResourceSyncRecordRepository syncRecordRepository,
                                AppEnv appEnv,
                                Times times) {
        this.dispatcher = DispatchInstance.of(instance);
        this.syncRecordRepository = syncRecordRepository;
        this.appEnv = appEnv;
        this.times = times;
        this.running = new AtomicBoolean(false);
    }


    @ConsumeEvent(value = SyncResourceFired.TOPIC, blocking = true)
    public void sync(SyncResourceFired event) {
        if (!running.compareAndSet(false, true)) {
            LOGGER.info("Resource Synchronizer is running.");
            return;
        }
        try {
            doSync(event);
        } finally {
            running.set(false);
        }
    }

    private void doSync(SyncResourceFired event) {
        SyncContext context = new SyncContext(event.getCourseId());
        QueueWorker.<SyncContext, String>builder()
                .name("Sync course resources")
                .onBefore(this::setupSyncRecord)
                .onPoll(c -> this.pollSyncRecord())
                .onWork((c, recordId) -> this.dispatchSync(recordId))
                .onFailed(this::onSyncFailed)
                .continueWhen((c, recordId, e) -> c.shouldContinue(recordId))
                .spinDuration(SPIN_DURATION)
                .build()
                .execute(context);
    }

    @Transactional(Transactional.TxType.REQUIRES_NEW)
    protected void setupSyncRecord(SyncContext context) {
        syncRecordRepository.resetByCourseId(context.getCourseId(), TimetableResourceSyncPriorities.MAX);
        syncRecordRepository.upsert(times.nowInstant(), TimetableResourceSyncPriorities.NEW);
        syncRecordRepository.resetByTimeout(times.nowInstant().minus(TIMEOUT), TimetableResourceSyncPriorities.PROCESSING_TIMEOUT);
        syncRecordRepository.resetByStatus(ProcessStatus.FAILED, TimetableResourceSyncPriorities.FAILED_TO_IDLE);
        syncRecordRepository.resetByStatus(ProcessStatus.POSTPONED, TimetableResourceSyncPriorities.POSTPONED_TO_IDLE);
    }

    @Transactional(Transactional.TxType.REQUIRES_NEW)
    protected void onSyncFailed(SyncContext context, String recordId, Exception e) {
        var entity = syncRecordRepository.lock(recordId);
        entity.setFailedMessage(Throwables.getStackTraceAsString(e));
        entity.setProcessStatus(ProcessStatus.FAILED);
        entity.setProcessEndAt(times.nowInstant());
        syncRecordRepository.persist(entity);
        context.markFailedRecord(recordId);
    }

    @Transactional(value = Transactional.TxType.REQUIRES_NEW)
    protected void dispatchSync(String recordId) {
        var entity = syncRecordRepository.lock(recordId);
        var enforceSucceedBefore = times.nowInstant().minus(ENFORCE_SUCCEED_DURATION);
        var syncSucceed = dispatcher.getOptional(appEnv.get())
                .map(synchronizer -> synchronizer.sync(entity))
                .filter(succeed -> succeed)
                .orElseGet(() -> entity.getTimetable().getEndAt().isAfter(enforceSucceedBefore));
        entity.setProcessStatus(syncSucceed ? ProcessStatus.SUCCEED : ProcessStatus.POSTPONED);
        entity.setProcessEndAt(times.nowInstant());
        syncRecordRepository.persist(entity);
    }

    @Transactional(Transactional.TxType.REQUIRES_NEW)
    protected String pollSyncRecord() {
        boolean findCompleted;
        Optional<TimetableResourceSyncRecordEntity> mayRecord;
        do {
            mayRecord = syncRecordRepository.peekAnyIdByDay()
                    .map(syncRecordRepository::lock);
            findCompleted = mayRecord.map(record -> record.getProcessStatus() == ProcessStatus.IDLE).orElse(true);
        } while (!findCompleted);
        mayRecord.ifPresent(record -> {
            record.setProcessStatus(ProcessStatus.PROCESSING);
            record.setProcessStartAt(times.nowInstant());
            record.setProcessEndAt(null);
            syncRecordRepository.persist(record);
        });
        return mayRecord
                .map(TimetableResourceSyncRecordEntity::getId)
                .orElse(null);
    }


    private static class SyncContext {
        private final Set<String> failedRecordIds;
        private final String courseId;

        public SyncContext(String courseId) {
            this.courseId = courseId;
            this.failedRecordIds = Sets.newConcurrentHashSet();
        }

        public void markFailedRecord(String id) {
            failedRecordIds.add(id);
        }

        public boolean shouldContinue(String id) {
            return !failedRecordIds.contains(id);
        }

        public String getCourseId() {
            return courseId;
        }
    }
}
